package com.zzg.common.message;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoPool;

public class KryoMessageConverter implements MessageConverter {
	private KryoPool kryoPool = null;

    public void setKryoPool(KryoPool kryoPool) {
        this.kryoPool = kryoPool;
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        Kryo kryo = null;
        com.zzg.common.base.message.Message object = null;
        try {
            kryo = kryoPool.borrow();
            kryo.register(com.zzg.common.base.message.Message.class);
            object = kryo.readObject(new Input(message.getBody()), com.zzg.common.base.message.Message.class);
        }
        finally {
            if(null != kryo) kryoPool.release(kryo);
        }
        return object;
    }

    @Override
    public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        Kryo kryo = null;
        Message message = null;
        try {
            kryo = kryoPool.borrow();
            Output o = new Output(1024, -1);
            kryo.writeObject(o, object);
            message = new Message(o.toBytes(), messageProperties);
        }
        finally {
            if(null != kryo) kryoPool.release(kryo);
        }
        return message;
    }

}
